package com.codejiwei.flink.table;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * author: codejiwei
 * date: 2023/8/7
 * desc:
 **/
public class Flink_Table_API_01 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
        //set time zone
        System.out.println(tEnv.getConfig().getLocalTimeZone().toString());
        tEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
        System.out.println(tEnv.getConfig().getLocalTimeZone().toString());

        tEnv.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("col1", DataTypes.STRING())
                        .column("col2", DataTypes.TIMESTAMP())
                        .column("col3", DataTypes.TIMESTAMP_LTZ())
                        .column("col4", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
                        .build())
                .option(DataGenConnectorOptions.NUMBER_OF_ROWS, 2L)
                .build());

        tEnv.executeSql("create temporary table SinkTable with ('connector' = 'print') like SourceTable(excluding options)");

        Table table2 = tEnv.from("SourceTable");

        Table table3 = tEnv.sqlQuery("select * from SourceTable");

        table3.executeInsert("SinkTable");
//        table2.executeInsert("SinkTable");


//        System.out.println(table2.explain());


    }


}
